package com.qiaofang.housewarehouse.dao.impl;

import com.qiaofang.housewarehouse.dao.HbaseDao;
import com.qiaofang.housewarehouse.dao.hbase.HBaseRow;
import com.qiaofang.housewarehouse.dao.hbase.HbaseClient;
import com.qiaofang.housewarehouse.dao.hbase.HbaseColumn;
import com.qiaofang.housewarehouse.stub.constants.TableConstants;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Repository;

import java.io.UnsupportedEncodingException;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.List;

@Repository
public class HbaseDaoImpl implements HbaseDao {
    private static final Logger logger = LoggerFactory.getLogger(HbaseDaoImpl.class);

    @Override
    public <T> List<T> listObjByRow(String tableName, String startRow, String stopRow, Class cls) throws Exception {
        List<T> all = new ArrayList<>();
        List<HBaseRow> list = HbaseClient.getResultScann(tableName, startRow, stopRow);
        for (HBaseRow row : list) {
            T obj = (T) cls.newInstance();
            Field fieldRowkey = cls.getDeclaredField("id");
            fieldRowkey.setAccessible(true);
            if (null != fieldRowkey) {
                fieldRowkey.set(obj, row.getRowKey());
            }
            for (HbaseColumn column : row.getDatas()) {
                String colName = column.getColName();
                String colValue = column.getColValue();
                Field fieldData = cls.getDeclaredField(colName);
                fieldData.setAccessible(true);
                if (null != fieldData) {
                    fieldData.set(obj, colValue);
                }
            }
            all.add(obj);
        }
        logger.info("遍历查询:{} Success!", tableName);
        return all;
    }

    @Override
    public <T> List<T> listObjByRowKeys(String tableName, List<String> rowKeys, Class cls) throws Exception {
        List<Result> list = HbaseClient.getRowKeys(tableName, rowKeys);
        List<T> all = new ArrayList<T>();
        for (Result rs : list) {
            if(!rs.isEmpty()){
                T obj = getObj(cls, rs);
                all.add(obj);
            }
        }
        return all;
    }

    @Override
    public <T> T getObjByRowKey(String tableName, String rowKey, Class cls) throws Exception {
        Result rs = HbaseClient.getRowKey(tableName, rowKey);
        if (!rs.isEmpty()) {
            T obj = getObj(cls, rs);
            return obj;
        } else {
            return null;
        }
    }

    private <T> T getObj(Class cls, Result rs) throws Exception {
        T obj = (T) cls.newInstance();
        Field fieldRowkey = cls.getDeclaredField("id");
        fieldRowkey.setAccessible(true);
        if (null != fieldRowkey) {
            fieldRowkey.set(obj, Bytes.toString(rs.getRow()));
        }
        for (Cell cell : rs.listCells()) {
            String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
            String value = Bytes.toString(CellUtil.cloneValue(cell));
            try {
                Field fieldData = cls.getDeclaredField(qualifier);
                fieldData.setAccessible(true);
                if (null != fieldData) {
                    fieldData.set(obj, value);
                }
            } catch (NoSuchFieldException e) {
                logger.info("{}中没有{},{}", cls.getName(),qualifier, e.getClass().getName());
            }
        }
        return obj;
    }

    @Override
    public <T> T getObjIncByRowKey(String tableName, String rowKey, Class cls) throws Exception {
        Result rs = HbaseClient.getRowKey(tableName, rowKey);
        if (null != rs.listCells()) {
            T obj = getObjInc(cls, rs);
            return obj;
        } else {
            return null;
        }
    }

    private <T> T getObjInc(Class cls, Result rs) throws InstantiationException, IllegalAccessException, NoSuchFieldException, UnsupportedEncodingException {
        T obj = (T) cls.newInstance();
        Field fieldRowkey = cls.getDeclaredField("id");
        fieldRowkey.setAccessible(true);
        if (null != fieldRowkey) {
            fieldRowkey.set(obj, new String(rs.getRow(), "utf-8"));
        }
        for (Cell cell : rs.listCells()) {
            String colName = new String(CellUtil.cloneQualifier(cell), "utf-8");
            Long value = Bytes.toLong(CellUtil.cloneValue(cell));
            Field fieldData = cls.getDeclaredField(colName);
            fieldData.setAccessible(true);
            if (null != fieldData) {
                fieldData.set(obj, value);
            }
        }
        return obj;
    }

    @Override
    public void insertObject(Object t, String tableName, String rowKey) throws Exception {
        List<HbaseColumn> cols = new ArrayList<>();
        Class cls = t.getClass();
        Field[] fields = cls.getDeclaredFields();
        for (Field f : fields) {
            f.setAccessible(true);
            if (!f.getName().equals("id")) {
                HbaseColumn col = new HbaseColumn();
                col.setFamilyName("info");
                col.setColName(f.getName());
                col.setColValue(f.get(t) == null ? "" : f.get(t).toString());
                cols.add(col);
            }
        }
        if (cols.size() > 0) {
            HbaseClient.insertRow(tableName, rowKey, cols);
        } else {
            logger.error("cols.size()={}", cols.size());
        }

    }

    @Override
    public <T> List<T> listObjectByStartEndRow(String tableName, String startRow, String endRow, Class cls, String type) throws Exception {
        List<T> all = new ArrayList<>();
        List<Result> list = HbaseClient.getResultListScann(tableName, startRow, endRow);
        for (Result rs : list) {
            T obj;
            if (type.equals(TableConstants.COLUMN_TYPE_LONG)) {
                obj = getObjInc(cls, rs);
            } else {
                obj = getObj(cls, rs);
            }
            all.add(obj);
        }
        return all;
    }

    @Override
    public <T> List<T> getResultsByRowFilter(String tableName, String rowkey, Class cls) throws Exception {
        List<T> all = new ArrayList<>();
        FilterList filters = new FilterList();
        Filter filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator(rowkey));
        filters.addFilter(filter);
        List<Result> results = HbaseClient.getResultsByFilter(tableName, filters);
        for (Result rs : results) {
            T obj = getObj(cls, rs);
            all.add(obj);
        }
        return all;
    }

    @Override
    public <T> List<T> getResultsByPrefixFilter(String tableName, String prefixRowkey, Class cls) throws Exception {
        List<T> all = new ArrayList<>();
        FilterList filters = new FilterList();
        Filter prefixFilter = new PrefixFilter(Bytes.toBytes(prefixRowkey));
        filters.addFilter(prefixFilter);
        List<Result> results = HbaseClient.getResultsByFilter(tableName, filters);
        for (Result rs : results) {
            T obj = getObj(cls, rs);
            all.add(obj);
        }
        return all;
    }
}
